package com.cet.recon.job;

import com.cet.recon.model.BankTxn;
import com.cet.recon.sftp.SftpDownloadTasklet;
import com.cet.recon.step.LoadCoreTxnTasklet;
import com.cet.recon.step.PublishTasklet;
import com.cet.recon.writer.ExcelReconItemWriter;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.batch.core.Job;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.configuration.annotation.JobScope;
import org.springframework.batch.core.configuration.annotation.StepScope;
import org.springframework.batch.core.job.builder.JobBuilder;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.core.step.builder.StepBuilder;
import org.springframework.batch.item.*;
import org.springframework.batch.item.database.JdbcBatchItemWriter;
import org.springframework.batch.item.database.JdbcPagingItemReader;
import org.springframework.batch.item.database.Order;
import org.springframework.batch.item.database.builder.JdbcBatchItemWriterBuilder;
import org.springframework.batch.item.database.support.MySqlPagingQueryProvider;
import org.springframework.batch.item.file.FlatFileItemReader;
import org.springframework.batch.item.file.builder.FlatFileItemReaderBuilder;
import org.springframework.batch.item.file.mapping.DefaultLineMapper;
import org.springframework.batch.item.file.transform.DelimitedLineTokenizer;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.io.FileSystemResource;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.transaction.PlatformTransactionManager;

import javax.sql.DataSource;
import java.math.BigDecimal;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.sql.Date;
import java.time.LocalDate;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;

/**
 * @author 30952
 */
@Configuration
public class ReconJobConfig {

  @Bean("reconJob")
  public Job reconJob(JobRepository repo,
                      @Qualifier("stepSftpFetch") Step stepSftpFetch,
                      @Qualifier("stepLoadBankFile") Step stepLoadBankFile,
                      @Qualifier("stepLoadCoreTxn") Step stepLoadCoreTxn,
                      @Qualifier("stepDiffMissingInBank") Step stepDiffMissingInBank,
                      @Qualifier("stepDiffMissingInCore") Step stepDiffMissingInCore,
                      @Qualifier("stepDiffMismatch") Step stepDiffMismatch,
                      @Qualifier("stepReport") Step stepReport,
                      @Qualifier("stepPublish") Step stepPublish) {
    return new JobBuilder("reconJob", repo)
        .start(stepSftpFetch)
        .next(stepLoadBankFile)
        .next(stepLoadCoreTxn)
        .next(stepDiffMissingInBank)
        .next(stepDiffMissingInCore)
        .next(stepDiffMismatch)
        .next(stepReport)
        .next(stepPublish)
        .build();
  }

  @Bean("stepSftpFetch")
  public Step stepSftpFetch(JobRepository repo, @Qualifier("resourcelessTransactionManager") PlatformTransactionManager tx, SftpDownloadTasklet t) {
    return new StepBuilder("stepSftpFetch", repo).tasklet(t, tx).build();
  }

  @Bean("stepLoadBankFile")
  public Step stepLoadBankFile(JobRepository repo,
                               @Qualifier("bizTransactionManager") PlatformTransactionManager tx,
                               @Qualifier("bankReader") FlatFileItemReader<BankTxn> bankReader,
                               @Qualifier("stgBankWriter") JdbcBatchItemWriter<BankTxn> stgBankWriter) {
    return new StepBuilder("stepLoadBankFile", repo)
        .<BankTxn, BankTxn>chunk(1000, tx)
        .reader(bankReader)
        .writer(stgBankWriter)
        .faultTolerant().skip(Exception.class).skipLimit(100)
        .build();
  }

  @Bean("stepLoadCoreTxn")
  public Step stepLoadCoreTxn(JobRepository repo,
                              @Qualifier("bizTransactionManager") PlatformTransactionManager tx,
                              LoadCoreTxnTasklet t) {
    return new StepBuilder("stepLoadCoreTxn", repo).tasklet(t, tx).build();
  }

    @Bean("stepDiffMissingInBank")
    public Step stepDiffMissingInBank(JobRepository repo,
                                      @Qualifier("bizTransactionManager") PlatformTransactionManager tx,
                                      @Qualifier("missingInBankReader") JdbcPagingItemReader<Map<String, Object>> missingInBankReader,
                                      @Qualifier("missingInBankWriter") JdbcBatchItemWriter<Map<String, Object>> missingInBankWriter) {
        return new StepBuilder("stepDiffMissingInBank", repo)
            .<Map<String, Object>, Map<String, Object>>chunk(1000, tx)
            .reader(missingInBankReader)
            .writer(missingInBankWriter)
            .build();
    }

    @Bean("stepDiffMissingInCore")
    public Step stepDiffMissingInCore(JobRepository repo,
                                      @Qualifier("bizTransactionManager") PlatformTransactionManager tx,
                                      @Qualifier("missingInCoreReader") JdbcPagingItemReader<Map<String, Object>> missingInCoreReader,
                                      @Qualifier("missingInCoreWriter") JdbcBatchItemWriter<Map<String, Object>> missingInCoreWriter) {
        return new StepBuilder("stepDiffMissingInCore", repo)
            .<Map<String, Object>, Map<String, Object>>chunk(1000, tx)
            .reader(missingInCoreReader)
            .writer(missingInCoreWriter)
            .build();
    }

    @Bean("stepDiffMismatch")
    public Step stepDiffMismatch(JobRepository repo,
                                 @Qualifier("bizTransactionManager") PlatformTransactionManager tx,
                                 @Qualifier("mismatchReader") JdbcPagingItemReader<Map<String, Object>> mismatchReader,
                                 @Qualifier("mismatchProcessor") ItemProcessor<Map<String, Object>, Map<String, Object>> mismatchProcessor,
                                 @Qualifier("mismatchWriter") JdbcBatchItemWriter<Map<String, Object>> mismatchWriter) {
        return new StepBuilder("stepDiffMismatch", repo)
            .<Map<String, Object>, Map<String, Object>>chunk(1000, tx)
            .reader(mismatchReader)
            .processor(mismatchProcessor)
            .writer(mismatchWriter)
            .build();
    }


    @Bean("stepReport")
    public Step stepReport(JobRepository repo,
                           @Qualifier("bizTransactionManager") PlatformTransactionManager tx,
                           @Qualifier("reconDiffReader") JdbcPagingItemReader<Map<String, Object>> reconDiffReader,
                           @Qualifier("excelReconWriter") ItemWriter<Map<String, Object>> excelReconWriter) {
        return new StepBuilder("stepReport", repo)
            .<Map<String, Object>, Map<String, Object>>chunk(1000, tx)
            .reader(reconDiffReader)
            .writer(excelReconWriter)
            .build();
    }


    @Bean
    public Step stepPublish(JobRepository repo,
                            @Qualifier("resourcelessTransactionManager") PlatformTransactionManager resTx,
                            PublishTasklet publishTasklet) {
        return new StepBuilder("stepPublish", repo)
            .tasklet(publishTasklet, resTx)
            .build();
    }


    @Bean("bankReader")
    @StepScope
    public FlatFileItemReader<BankTxn> bankReader(
        @Value("#{jobParameters['run.date']}") String runDate) {

        String filename = "work/inbox/recon_" + runDate + ".csv";

        DefaultLineMapper<BankTxn> mapper = new DefaultLineMapper<>();
        DelimitedLineTokenizer tok = new DelimitedLineTokenizer();
        tok.setNames("txDate","accountNo","orderNo","amount","currency","counterpart","txType");
        mapper.setLineTokenizer(tok);
        mapper.setFieldSetMapper(fs -> {
            BankTxn t = new BankTxn();
            t.setTxDate(LocalDate.parse(fs.readString("txDate")));
            t.setAccountNo(fs.readString("accountNo"));
            t.setOrderNo(fs.readString("orderNo"));
            t.setAmount(fs.readBigDecimal("amount"));
            t.setCurrency(fs.readString("currency"));
            t.setCounterpart(fs.readString("counterpart"));
            t.setTxType(fs.readString("txType"));
            t.setRawLine(String.join(",", fs.getValues()));
            return t;
        });

        return new FlatFileItemReaderBuilder<BankTxn>()
            .name("bankFileReader")
            .resource(new FileSystemResource(filename))
            .linesToSkip(1)
            .lineMapper(mapper)
            .build();
    }

    @Bean("stgBankWriter")
    public JdbcBatchItemWriter<BankTxn> stgBankWriter(@Qualifier("bizDataSource") DataSource ds) {
        return new JdbcBatchItemWriterBuilder<BankTxn>()
            .dataSource(ds)
            .sql("INSERT INTO stg_bank_txn (tx_date,account_no,order_no,amount,currency,counterpart,tx_type,raw_line) " +
                "VALUES (?,?,?,?,?,?,?,?) ON DUPLICATE KEY UPDATE amount=VALUES(amount),currency=VALUES(currency)," +
                "counterpart=VALUES(counterpart),tx_type=VALUES(tx_type),raw_line=VALUES(raw_line)")
            .itemPreparedStatementSetter((it, ps) -> {
                ps.setDate(1, Date.valueOf(it.getTxDate()));
                ps.setString(2, it.getAccountNo());
                ps.setString(3, it.getOrderNo());
                ps.setBigDecimal(4, it.getAmount());
                ps.setString(5, it.getCurrency());
                ps.setString(6, it.getCounterpart());
                ps.setString(7, it.getTxType());
                ps.setString(8, it.getRawLine());
            })
            .build();
    }

    @Bean("missingInBankReader")
    @StepScope
    public JdbcPagingItemReader<Map<String, Object>> missingInBankReader(
        @Qualifier("bizDataSource") DataSource dataSource,
        @Value("#{jobParameters['run.date']}") String runDate) {

        MySqlPagingQueryProvider qp = new MySqlPagingQueryProvider();
        qp.setSelectClause("c.tx_date, c.account_no, c.order_no, c.amount, c.currency, c.tx_type");
        qp.setFromClause("from stg_core_txn c left join stg_bank_txn b " +
            "on b.tx_date=c.tx_date and b.account_no=c.account_no and b.order_no=c.order_no");
        qp.setWhereClause("where c.tx_date = :runDate and b.order_no is null");
        qp.setSortKeys(Map.of("account_no", Order.ASCENDING, "order_no", Order.ASCENDING));

        JdbcPagingItemReader<Map<String, Object>> reader = new JdbcPagingItemReader<>();
        reader.setDataSource(dataSource);
        reader.setPageSize(1000);
        reader.setQueryProvider(qp);
        Map<String, Object> params = new HashMap<>();
        params.put("runDate", java.sql.Date.valueOf(runDate));
        reader.setParameterValues(params);
        reader.setRowMapper((rs, rowNum) -> {
            Map<String, Object> m = new HashMap<>();
            m.put("tx_date", rs.getDate("tx_date"));
            m.put("account_no", rs.getString("account_no"));
            m.put("order_no", rs.getString("order_no"));
            m.put("amount", rs.getBigDecimal("amount"));
            m.put("currency", rs.getString("currency"));
            m.put("tx_type", rs.getString("tx_type"));
            return m;
        });
        return reader;
    }

    @Bean("missingInBankWriter")
    public JdbcBatchItemWriter<Map<String, Object>> missingInBankWriter(@Qualifier("bizDataSource") DataSource ds) {
        return new JdbcBatchItemWriterBuilder<Map<String, Object>>()
            .dataSource(ds)
            .sql("INSERT INTO recon_diff " +
                "(id, tx_date, account_no, order_no, diff_type, core_amount, core_currency, core_tx_type, created_at) " +
                "VALUES (UUID(), :tx_date, :account_no, :order_no, 'MISSING_IN_BANK', :amount, :currency, :tx_type, NOW())")
            .columnMapped()
            .build();
    }
    @Bean("missingInCoreReader")
    @StepScope
    public JdbcPagingItemReader<Map<String, Object>> missingInCoreReader(
        @Qualifier("bizDataSource") DataSource dataSource,
        @Value("#{jobParameters['run.date']}") String runDate) {

        MySqlPagingQueryProvider qp = new MySqlPagingQueryProvider();
        qp.setSelectClause("b.tx_date, b.account_no, b.order_no, b.amount, b.currency, b.tx_type");
        qp.setFromClause("from stg_bank_txn b left join stg_core_txn c " +
            "on b.tx_date=c.tx_date and b.account_no=c.account_no and b.order_no=c.order_no");
        qp.setWhereClause("where b.tx_date = :runDate and c.order_no is null");
        qp.setSortKeys(Map.of("account_no", Order.ASCENDING, "order_no", Order.ASCENDING));

        JdbcPagingItemReader<Map<String, Object>> reader = new JdbcPagingItemReader<>();
        reader.setDataSource(dataSource);
        reader.setPageSize(1000);
        reader.setQueryProvider(qp);
        Map<String, Object> params = new HashMap<>();
        params.put("runDate", java.sql.Date.valueOf(runDate));
        reader.setParameterValues(params);
        reader.setRowMapper((rs, rowNum) -> {
            Map<String, Object> m = new HashMap<>();
            m.put("tx_date", rs.getDate("tx_date"));
            m.put("account_no", rs.getString("account_no"));
            m.put("order_no", rs.getString("order_no"));
            m.put("amount", rs.getBigDecimal("amount"));
            m.put("currency", rs.getString("currency"));
            m.put("tx_type", rs.getString("tx_type"));
            return m;
        });
        return reader;
    }

    @Bean("missingInCoreWriter")
    @StepScope
    public JdbcBatchItemWriter<Map<String, Object>> missingInCoreWriter(@Qualifier("bizDataSource") DataSource ds) {
        return new JdbcBatchItemWriterBuilder<Map<String, Object>>()
            .dataSource(ds)
            .sql("INSERT INTO recon_diff " +
                "(id, tx_date, account_no, order_no, diff_type, bank_amount, bank_currency, bank_tx_type, created_at) " +
                "VALUES (UUID(), :tx_date, :account_no, :order_no, 'MISSING_IN_CORE', :amount, :currency, :tx_type, NOW())")
            .columnMapped()
            .build();
    }

    @Bean("mismatchReader")
    @StepScope
    public JdbcPagingItemReader<Map<String, Object>> mismatchReader(
        @Qualifier("bizDataSource") DataSource dataSource,
        @Value("#{jobParameters['run.date']}") String runDate) {

        MySqlPagingQueryProvider qp = new MySqlPagingQueryProvider();
        qp.setSelectClause("c.tx_date, c.account_no, c.order_no, " +
            "c.amount as core_amount, b.amount as bank_amount, " +
            "c.currency as core_currency, b.currency as bank_currency, " +
            "c.tx_type as core_tx_type, b.tx_type as bank_tx_type, " +
            "c.counterpart as core_counterpart, b.counterpart as bank_counterpart");
        qp.setFromClause("from stg_core_txn c join stg_bank_txn b " +
            "on b.tx_date=c.tx_date and b.account_no=c.account_no and b.order_no=c.order_no");
        qp.setWhereClause("where c.tx_date = :runDate");
        qp.setSortKeys(Map.of("account_no", Order.ASCENDING, "order_no", Order.ASCENDING));

        JdbcPagingItemReader<Map<String, Object>> reader = new JdbcPagingItemReader<>();
        reader.setDataSource(dataSource);
        reader.setPageSize(1000);
        reader.setQueryProvider(qp);
        Map<String, Object> params = new HashMap<>();
        params.put("runDate", java.sql.Date.valueOf(runDate));
        reader.setParameterValues(params);
        reader.setRowMapper((rs, rowNum) -> {
            Map<String, Object> m = new HashMap<>();
            m.put("tx_date", rs.getDate("tx_date"));
            m.put("account_no", rs.getString("account_no"));
            m.put("order_no", rs.getString("order_no"));
            m.put("core_amount", rs.getBigDecimal("core_amount"));
            m.put("bank_amount", rs.getBigDecimal("bank_amount"));
            m.put("core_currency", rs.getString("core_currency"));
            m.put("bank_currency", rs.getString("bank_currency"));
            m.put("core_tx_type", rs.getString("core_tx_type"));
            m.put("bank_tx_type", rs.getString("bank_tx_type"));
            m.put("core_counterpart", rs.getString("core_counterpart"));
            m.put("bank_counterpart", rs.getString("bank_counterpart"));
            return m;
        });
        return reader;
    }

    @Bean("mismatchProcessor")
    @StepScope
    public ItemProcessor<Map<String, Object>, Map<String, Object>> mismatchProcessor(
        @Value("${recon.amount-tolerance}") BigDecimal tol) {
        return item -> {
            BigDecimal coreAmt = (BigDecimal) item.get("core_amount");
            BigDecimal bankAmt = (BigDecimal) item.get("bank_amount");
            String coreCcy = (String) item.get("core_currency");
            String bankCcy = (String) item.get("bank_currency");
            String coreType = (String) item.get("core_tx_type");
            String bankType = (String) item.get("bank_tx_type");
            String coreCp = (String) item.get("core_counterpart");
            String bankCp = (String) item.get("bank_counterpart");

            boolean amtDiff = coreAmt != null && bankAmt != null
                && coreAmt.subtract(bankAmt).abs().compareTo(tol) > 0;
            boolean fieldDiff = !Objects.equals(coreCcy, bankCcy)
                || !Objects.equals(coreType, bankType)
                || !Objects.equals(nullToEmpty(coreCp), nullToEmpty(bankCp));

            if (!amtDiff && !fieldDiff) {
                return null; // 不需要写差异
            }

            Map<String, Object> out = new HashMap<>(item);
            out.put("diff_type", amtDiff ? "AMOUNT_MISMATCH" : "FIELD_MISMATCH");
            return out;
        };
    }

    @Bean("mismatchWriter")
    public JdbcBatchItemWriter<Map<String, Object>> mismatchWriter(@Qualifier("bizDataSource") DataSource ds) {
        return new JdbcBatchItemWriterBuilder<Map<String, Object>>()
            .dataSource(ds)
            .sql("INSERT INTO recon_diff " +
                "(id, tx_date, account_no, order_no, diff_type, " +
                " core_amount, bank_amount, core_currency, bank_currency, core_tx_type, bank_tx_type, extra, created_at) " +
                "VALUES (UUID(), :tx_date, :account_no, :order_no, :diff_type, " +
                " :core_amount, :bank_amount, :core_currency, :bank_currency, :core_tx_type, :bank_tx_type, NULL, NOW())")
            .columnMapped()
            .build();
    }


    private static String nullToEmpty(String s) { return s == null ? "" : s; }


    @Bean("reconDiffReader")
    @StepScope
    public JdbcPagingItemReader<Map<String, Object>> reconDiffReader(
        @Qualifier("bizDataSource") DataSource ds,
        @Value("#{jobParameters['run.date']}") String runDate) {

        MySqlPagingQueryProvider qp = new MySqlPagingQueryProvider();
        qp.setSelectClause("id, tx_date, account_no, order_no, diff_type, core_amount, bank_amount, core_currency, bank_currency");
        qp.setFromClause("from recon_diff");
        qp.setWhereClause("where tx_date = :runDate");
        qp.setSortKeys(Map.of("order_no", Order.ASCENDING));

        JdbcPagingItemReader<Map<String, Object>> reader = new JdbcPagingItemReader<>();
        reader.setDataSource(ds);
        reader.setQueryProvider(qp);
        reader.setPageSize(1000);
        Map<String, Object> params = new HashMap<>();
        params.put("runDate", java.sql.Date.valueOf(runDate));
        reader.setParameterValues(params);
        reader.setRowMapper((rs, rn) -> {
            Map<String, Object> m = new HashMap<>();
            m.put("tx_date", rs.getDate("tx_date"));
            m.put("account_no", rs.getString("account_no"));
            m.put("order_no", rs.getString("order_no"));
            m.put("diff_type", rs.getString("diff_type"));
            m.put("core_amount", rs.getBigDecimal("core_amount"));
            m.put("bank_amount", rs.getBigDecimal("bank_amount"));
            m.put("core_currency", rs.getString("core_currency"));
            m.put("bank_currency", rs.getString("bank_currency"));
            return m;
        });
        return reader;
    }

    @Bean("excelReconWriter")
    @StepScope
    public ItemWriter<Map<String, Object>> excelReconWriter(
        @Value("#{jobParameters['run.date']}") String runDate) {
        return new ExcelReconItemWriter(
            Paths.get("work", "report", "recon_diff_" + runDate + ".xlsx")
        );
    }

    //transactionManger使用注意点

    // step中的reader, processor, writer中包含对业务库的写操作，则使用bizTransactionManager,(Reader, Writer, Processor中同理）
    // JobRepository使用@BatchDatasource指定的数据源
    // 其他场景可以直接使用resourcelessTransactionManager

}
